package com.arch.flink.archetype.network;

import com.arch.flink.archetype.network.handler.OutboundEnvelopeEncoder;
import org.apache.flink.shaded.netty4.io.netty.bootstrap.Bootstrap;
import org.apache.flink.shaded.netty4.io.netty.buffer.PooledByteBufAllocator;
import org.apache.flink.shaded.netty4.io.netty.channel.Channel;
import org.apache.flink.shaded.netty4.io.netty.channel.ChannelFuture;
import org.apache.flink.shaded.netty4.io.netty.channel.ChannelFutureListener;
import org.apache.flink.shaded.netty4.io.netty.channel.ChannelInitializer;
import org.apache.flink.shaded.netty4.io.netty.channel.ChannelOption;
import org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoopGroup;
import org.apache.flink.shaded.netty4.io.netty.channel.socket.SocketChannel;
import org.apache.flink.shaded.netty4.io.netty.channel.socket.nio.NioSocketChannel;

import java.net.InetSocketAddress;
import java.util.concurrent.TimeUnit;

/**
 * @author pizhihui
 * @date 2024-05-20 18:02
 */
public class NettyClient {

    private Bootstrap out;

    private final int numInThreads = 10;


    private final int numOutThreads = 10;
    private final int bufferSize = 2048;


    private final int lowWaterMark;
    private final int highWaterMark;


    public NettyClient() {
        int lowWaterMark = 1024;
        int highWaterMark = 2048;
        this.lowWaterMark = (lowWaterMark == -1) ? bufferSize / 2 : lowWaterMark;
        this.highWaterMark = (highWaterMark == -1) ? bufferSize : highWaterMark;
    }

    public void start() {

        int numHeapArenas = 0;
        int numDirectArenas = numInThreads + numOutThreads;
        int pageSize = bufferSize << 1;
        int chunkSize = 16 * 1 << 20; // 16 MB
        int maxOrder = (int) (Math.log(chunkSize/pageSize) / Math.log(2));


        PooledByteBufAllocator pooledByteBufAllocator =
                new PooledByteBufAllocator(true, numHeapArenas, numDirectArenas, pageSize, maxOrder);



        // --------------------------------------------------------------------
        // client bootstrap (outgoing connections)
        // --------------------------------------------------------------------
        out = new Bootstrap();
        out.group(new NioEventLoopGroup(numOutThreads))
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel channel) throws Exception {
                        channel.pipeline()
                                .addLast(new OutboundEnvelopeEncoder());
                    }
                })
                .option(ChannelOption.WRITE_BUFFER_LOW_WATER_MARK, lowWaterMark)
                .option(ChannelOption.WRITE_BUFFER_HIGH_WATER_MARK, highWaterMark)
                .option(ChannelOption.ALLOCATOR, pooledByteBufAllocator)
                .option(ChannelOption.TCP_NODELAY, false)
                .option(ChannelOption.SO_KEEPALIVE, true);
    }


    public void write() {
        this.out.connect(new InetSocketAddress("127.0.0.1", 4096))
                .addListener(new ChannelInBuildup());
    }

    private static final class ChannelInBuildup implements ChannelFutureListener {


        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            if (future.isSuccess()) {

                Channel channel = future.channel();
                System.out.println("当前的 channel: " + channel);

                // channel.writeAndFlush("hello jajva");

            }
        }
    }

    public static void main(String[] args) throws InterruptedException {

        NettyClient client = new NettyClient();
        client.start();


        // TimeUnit.SECONDS.sleep(10);


        client.write();


    }

}
